/*
 * Copyright 2009 Brian Mackay brianwmackay@msn.com
 *   All Rights Reserved. See end of file
 *
 *  Large portions of this code were taken from Hadoop Core
 */
package com.bmackay.vonsteuben.core.hdfs;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.fs.FileUtil;

public class CopyTasks {

    private static Path checkDest(String srcName, FileSystem dstFS, Path dst,
            boolean overwrite) throws IOException {

        if (dstFS.exists(dst)) {
            FileStatus sdst = dstFS.getFileStatus(dst);
            if (sdst.isDir()) {
                if (null == srcName) {
                    throw new IOException("Target " + dst + " is a directory");
                }
                return checkDest(null, dstFS, new Path(dst, srcName), overwrite);
            } else if (!overwrite) {
                throw new IOException("Target " + dst + " already exists");
            }
        }
        return dst;
    }


    public static void copyMergeToLocal(String srcf, Path dst, boolean endline, String filter, boolean overwrite) throws IOException {

        Path srcPath = new Path(srcf);
        FileSystem srcFs = srcPath.getFileSystem(new Configuration());
        Path[] srcs = FileUtil.stat2Paths(srcFs.globStatus(srcPath), srcPath);

        for (int i = 0; i < srcs.length; i++) {

            if (endline) {
                copyMerge(srcFs, srcs[i],
                        FileSystem.getLocal(new Configuration()), dst,
                        false, new Configuration(), "\n", filter, overwrite);
            } else {
                copyMerge(srcFs, srcs[i],
                        FileSystem.getLocal(new Configuration()), dst,
                        false, new Configuration(), null, filter, overwrite);
            }
        }
    }

    public static boolean copyMerge(FileSystem srcFS, Path srcDir,
            FileSystem dstFS, Path dstFile,
            boolean deleteSource,
            Configuration conf, String addString, String filter, boolean overwrite) throws IOException {


        String newName = "";
        if (!srcDir.getName().contains(".txt")) {
            newName = srcDir.getName() + ".txt";
            dstFile = checkDest(newName, dstFS, dstFile, overwrite);
        } else {
            dstFile = checkDest(srcDir.getName(), dstFS, dstFile, overwrite);
        }


        dstFile = checkDest(srcDir.getName(), dstFS, dstFile, overwrite);
        System.out.print("dstFile a = " + dstFile.toString() + " \n");

        if (!srcFS.getFileStatus(srcDir).isDir()) {
            return false;
        }

        OutputStream out = dstFS.create(dstFile);

        try {
            FileStatus contents[] = srcFS.listStatus(srcDir);
            for (int i = 0; i < contents.length; i++) {
                String fileTocopy = "";
                boolean isSkipped = false;
                if (!contents[i].getPath().getName().contains(filter)) {
                    isSkipped = true;
                }
                if (isSkipped == false) {
                    fileTocopy = contents[i].getPath().getName();
                    System.out.print("fileTocopy = " + fileTocopy + "\n");
                    if (!contents[i].isDir()) {

                        InputStream in = srcFS.open(contents[i].getPath());
                        try {
                            IOUtils.copyBytes(in, out, conf, false);
                            if (addString != null) {
                                out.write(addString.getBytes("UTF-8"));
                            }

                        } finally {
                            in.close();
                        }
                    }
                }
            }
        } finally {
            out.close();
        }

        if (deleteSource) {
            return srcFS.delete(srcDir, true);
        } else {
            return true;
        }
    }

}

/*
 * Copyright 2009 Brian Mackay brianwmackay@msn.com
 * All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the 'License');
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an 'AS IS' BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
